package com.weizhu.swufer.core.rpc;

import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.weizhu.swufer.constant.SwuferConstant;
import com.weizhu.swufer.core.entity.RpcBaseInfo;
import com.weizhu.swufer.core.grpc.RemoteCacheManager;
import com.weizhu.swufer.core.grpc.GrpcServer;
import com.weizhu.swufer.core.monitor.RpcMonitor;
import com.weizhu.swufer.core.registry.ServiceRegister;
import com.weizhu.swufer.core.registry.ZkServiceRegister;

import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class RpcEngine {

    private ServiceRegister register;

    private GrpcServer server;

    private ConcurrentHashMap<String,RpcBaseInfo> localServiceMap;

    private Properties properties;

    private volatile static RpcEngine instance;

    private RemoteCacheManager channelManager;

    private ListeningExecutorService asyncThreadPool;

    private RpcMonitor rpcMonitor;

    private static AtomicInteger barrier = new AtomicInteger(0);

    private RpcEngine(){

    }
    
    public static RpcEngine getInstance(){
        if (instance == null){
            synchronized (RpcEngine.class){
                if (instance == null){
                    instance = new RpcEngine();
                }
            }
        }
        return instance;
    }

    public void initEngine(Properties properties){
        localServiceMap = new ConcurrentHashMap<>();
        register = new ZkServiceRegister(properties);
        channelManager = RemoteCacheManager.getInstance();
        this.properties = properties;
        initAsyncThreadPool();
        rpcMonitor = RpcMonitor.getInstance();
    }

    private void initAsyncThreadPool() {
        if (asyncThreadPool == null){
            asyncThreadPool = MoreExecutors.listeningDecorator(
                    Executors.newFixedThreadPool(32)
            );
        }
    }

    public ListeningExecutorService getAsyncThreadPool() {
        return asyncThreadPool;
    }

    public void setAsyncThreadPool(ListeningExecutorService asyncThreadPool) {
        this.asyncThreadPool = asyncThreadPool;
    }

    public ServiceRegister getRegister() {
        return register;
    }

    public void setRegister(ServiceRegister register) {
        this.register = register;
    }

    public GrpcServer getServer() {
        return server;
    }

    public void setServer(GrpcServer server) {
        this.server = server;
    }

    public void registerRpcService(List<RpcBaseInfo> rpcBaseInfoList) {

        Timer timer = new Timer();
        TimerTask timerTask =new TimerTask(){
            @Override
            public void run() {
                registerInterface(rpcBaseInfoList);
                registerInRegisterCenter(rpcBaseInfoList);

            }
        };
        timer.schedule(timerTask,Long.parseLong(properties.getProperty(SwuferConstant.PROPERTIES_LAZY_START)));



    }

    private void registerInRegisterCenter(List<RpcBaseInfo> rpcBaseInfoList){
        register.register(rpcBaseInfoList);

    }

    private void registerInterface(List<RpcBaseInfo> rpcBaseInfoList){
        for (RpcBaseInfo rpcBaseInfo : rpcBaseInfoList) {
            localServiceMap.putIfAbsent(rpcBaseInfo.getService(),rpcBaseInfo);
            System.out.println(rpcBaseInfo.getService() + " has been registered");
        }
    }

    /**
     * rpc引擎启动，初始化并开启grpc服务器
     * @throws IOException
     */
    public void start() throws IOException {
        this.server = new GrpcServer(properties);
        this.server.initServer(localServiceMap);
        this.server.start();
    }

    public ConcurrentHashMap<String, RpcBaseInfo> getLocalServiceMap() {
        return localServiceMap;
    }

    public void setLocalServiceMap(ConcurrentHashMap<String, RpcBaseInfo> localServiceMap) {
        this.localServiceMap = localServiceMap;
    }

    public AtomicInteger getBarrier() {
        return barrier;
    }
}
